%%%-------------------------------------------------------------------
%%% File    : svc_rpool.erl
%%% Author  :  <vjache>
%%% Description : Resource Pool Service. It is designed to provide fast 
%%%				  access to the multiple resources for which:
%%%					* open & close operations have a big enough costs
%%%					* the number of possible opened resources is 
%%%					  restricted by some factor (e.g. file handles by OS)
%%%				  This server has a cache in which all opened resources 
%%%				  are stored, hence this service could be called as a
%%%				  Resource Cache Service.
%%%				  The nature of resource is introduced by resource factory
%%%				  which is aware about how to:
%%%					* open
%%%					* close
%%%					* destroy
%%%				  a resource. Service actually responsible for resource
%%%				  life cycle controll and cleanup.
%%%
%%% Created : 30 Jun 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(svc_rpool).

-behaviour(gen_server).

-include("../include/util.hrl").
-include("../include/rpool.hrl").

-include_lib("eunit/include/eunit.hrl").

%% API
-export([start_link/4,
		 start_link/1,
		 stop/1,
		 open/2,
		 close/2,
		 close_finalize/2,
		 destroy/2,
		 short_action/3,
		 long_action/3]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
	 terminate/2, code_change/3]).

-define(SERVER,?MODULE).
-define(ENTRY(ResourceId,Timestamp,UsageCount,ResourceHandle),
		{ResourceId,Timestamp,UsageCount,ResourceHandle}).
-define(CHECKERR(Expr,ErrTag),checkerr(Expr,ErrTag)).

-record(state, {name,factory,cache,cache_max_size,clean_factor=0.5}).

%%====================================================================
%% API
%%====================================================================

%%--------------------------------------------------------------------
%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
%% Description: Starts the server
%%--------------------------------------------------------------------
start_link({PoolName,CacheMaxSize,CleanFactor,{M,F,A}=_ResourceFactoryProvider}) ->
	start_link(PoolName,CacheMaxSize,CleanFactor,apply(M,F,A)).
start_link(PoolName,CacheMaxSize,CleanFactor,ResourceFactory) when 
  is_atom(PoolName),is_record(ResourceFactory, rsc_factory) ->
    gen_server:start_link({local, PoolName}, ?MODULE, [PoolName,CacheMaxSize,CleanFactor,ResourceFactory], []).
stop(PoolName) ->
	srv_call(PoolName,stop, stop_err).


open(PoolName, ResourceId) ->
	srv_call(PoolName,{open, ResourceId, 1},open_err).
close_finalize(PoolName, ResourceId) ->
	srv_call(PoolName,{close, ResourceId, true},close_finalize_err).
close(PoolName, ResourceId) ->
	srv_call(PoolName,{close, ResourceId, false},close_err).
destroy(PoolName, ResourceId) ->
	srv_call(PoolName,{destroy, ResourceId},destroy_err).
short_action(PoolName, ResourceId, DoFun) ->
	srv_call(PoolName,{short_action, ResourceId, DoFun},short_action_err).
long_action(PoolName, ResourceId, DoFun) ->
	ActionPid=srv_call(PoolName,{long_action, ResourceId, DoFun},long_action_err),
	receive
		{ok,ActionPid,Result} ->
			Result;
		{error,ActionPid,Err} ->
			throw(Err)
	end.

%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% Function: init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([PoolName,CacheMaxSize,CleanFactor,ResourceFactory]) ->
	CacheId=list_to_atom((atom_to_list(PoolName)++"_cache")),
    {ok, #state{
				name=PoolName,
				factory=ResourceFactory,
				cache_max_size=CacheMaxSize,
				clean_factor=CleanFactor,
				cache=ets:new(CacheId, [named_table,set,protected,{keypos,1}])}}.

%%--------------------------------------------------------------------
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call(stop, _From, State) ->
	{stop, shutdown , ok, State};
handle_call(Request, From, State) ->
    try call({Request, From, State}) of
		{Reply, State1} ->
			{reply, Reply, State1}
	catch
		throw:Reason ->
			{reply, {error,{Reason,at,erlang:get_stacktrace()}}, State}
	end.

call({Request={open, ResourceId,Incr}, {CallerPid,_Tag}=From, State}) 
  when Incr==0 orelse Incr==1 ->
	CacheId=State#state.cache,
	Factory=State#state.factory,
	OpenFun=Factory#rsc_factory.open_fun,
	case ets:lookup(CacheId, ResourceId) of
		[] ->
			CacheSize=ets:info(CacheId, size),
			MaxSize=State#state.cache_max_size,
			if
				MaxSize==infinity orelse 
					CacheSize<MaxSize ->
					try OpenFun(ResourceId) of
						RscHandle when Incr==1 ->
							true=ets:insert_new(CacheId, ?ENTRY(ResourceId,now(),Incr,RscHandle)),
							start_monitor(State, CallerPid, ResourceId),
							{ok,State};
						RscHandle when Incr==0 ->
							true=ets:insert_new(CacheId, ?ENTRY(ResourceId,now(),Incr,RscHandle)),
							{ok,State}
					catch
						throw:limit_err ->
							gc_pool(State),
							call({Request, From, State});
						_:Err ->
							throw({open_rsc_err,ResourceId,Err})
					end;
				CacheSize>=MaxSize ->
					gc_pool(State),
					call({Request, From, State})
			end;
		[?ENTRY(ResourceId,_,Count,_)] when Incr==1 ->
			ets:update_element(CacheId,ResourceId,[{2,now()},{3,Count+1}]),
			start_monitor(State, CallerPid, ResourceId),
			{ok,State};
		[?ENTRY(ResourceId,_,_,_)] when Incr==0 ->
			ets:update_element(CacheId, ResourceId,{2,now()}),
			{ok,State}		
	end;
call({_Request={close, ResourceId, Finalize}, _From, State}) ->
	CacheId=State#state.cache,
	try ets:lookup_element(CacheId, ResourceId, 3) of
		0 ->
			finalize_resource(State, ResourceId),
			{ok,State};
		Cnt when Cnt<0 ->
			throw({cache_count_underflow,{ResourceId}});
		Cnt when Cnt>0  ->
			case ets:update_counter(CacheId, ResourceId, {3,-1}) of
				0 when Finalize==true ->
					finalize_resource(State, ResourceId);
				_ ->
					noop
			end,
			{ok,State}
	catch
		_:badarg -> throw({resource_not_open,ResourceId})
	end;
call({_Request={destroy, ResourceId}, _From, State}) ->
	CacheId=State#state.cache,
	Factory=State#state.factory,
	DestroyFun=Factory#rsc_factory.destroy_fun,
	try ets:lookup_element(CacheId, ResourceId, 3) of
		0 ->
			finalize_resource(State, ResourceId),
			DestroyFun(ResourceId),
			{ok,State};
		_ ->
			throw({resource_in_use,ResourceId})
	catch
		_:badarg -> throw({resource_not_open,ResourceId})
	end;
call({_Request={short_action, ResourceId, DoFun}, From, State}) ->
	{ok, State1}=call({{open, ResourceId,0}, From, State}),
	try DoFun(get_handle(State,ResourceId)) of
		Result ->
			{Result, State1}
	catch
		_:Err ->
			{{error,Err},State1}
	end;
call({_Request={long_action, ResourceId, DoFun}, {CallerPid,_Tag}=_From, State}) ->
%% 	CacheId=State#state.cache,
	ActionPid=spawn_link(
	  fun() ->
			  try open(State#state.name,ResourceId),
				  DoFun(get_handle(State, ResourceId)) of
				  Result ->
					  CallerPid ! {ok,self(),Result}
			  catch
				  _:Error ->
					  CallerPid ! {error,self(),Error}
			  end
	  end),
	{ActionPid,State}.

get_handle(State,ResourceId) ->
	CacheId=State#state.cache,
	ets:lookup_element(CacheId, ResourceId, 4).
get_count(State,ResourceId) ->
	CacheId=State#state.cache,
	ets:lookup_element(CacheId, ResourceId, 3).

gc_pool(State) ->
	CacheId=State#state.cache,
	Factory=State#state.factory,
	CloseFun=Factory#rsc_factory.close_fun,
	GItems=ets:foldl(
	  fun(?ENTRY(_ResourceId,_Timestamp,Count,_Handle)=Item,AccIn)->
			  case Count of
				  0 ->
					  [Item|AccIn];
				  _ ->
					  AccIn
			  end
	  end, [], CacheId),
	HowMuchClean=round(length(GItems)*State#state.clean_factor),
	GItemsByTs=lists:keysort(2,GItems),
	case gc_loop(CacheId,CloseFun,GItemsByTs,HowMuchClean,0) of
		0 ->
			throw(pool_full);
		CleanedCount ->
			CleanedCount
	end.

gc_loop(_,_,[],_,FinalizedCount) ->
	FinalizedCount;
gc_loop(_,_,_,0,FinalizedCount) ->
	FinalizedCount;
gc_loop(CacheId,CloseFun,
	   [?ENTRY(ResourceId,_Timestamp,_Count,_Handle)|Tail], Max, FinalizedCount) ->
	finalize_resource(CacheId, ResourceId, CloseFun),
	gc_loop(CacheId,CloseFun,Tail,Max-1,FinalizedCount+1).

finalize_resource(State, ResourceId) when is_record(State, state) ->
	CacheId=State#state.cache,
	Factory=State#state.factory,
	CloseFun=Factory#rsc_factory.close_fun,
	finalize_resource(CacheId, ResourceId, CloseFun).
finalize_resource(CacheId, ResourceId, CloseFun) when is_function(CloseFun,1) ->
	Handle=ets:lookup_element(CacheId, ResourceId, 4),
	CloseFun(Handle),
	ets:delete(CacheId, ResourceId).
	
start_monitor(State, Pid, ResourceId) ->
	spawn_link(
	  fun() ->
			  MonRef=erlang:monitor(process, Pid),
			  monitor_loop(State, Pid, ResourceId, MonRef)
	  end).
monitor_loop(State, Pid, ResourceId, MonRef) ->
	receive
		{'DOWN', MonRef, process, Pid, normal} ->
			ok;
		{'DOWN', MonRef, process, Pid, _Reason} ->
			(catch close(State#state.name,ResourceId))
	after
			10000 ->
			try get_count(State, ResourceId) of
				0 ->
					ok;
				_ ->
					monitor_loop(State, Pid, ResourceId, MonRef)
			catch
				_:badarg ->
					ok
			end
	end.

%%--------------------------------------------------------------------
%% Function: handle_cast(Msg, State) -> {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, State}
%% Description: Handling cast messages
%%--------------------------------------------------------------------
handle_cast(_Msg, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: handle_info(Info, State) -> {noreply, State} |
%%                                       {noreply, State, Timeout} |
%%                                       {stop, Reason, State}
%% Description: Handling all non call/cast messages
%%--------------------------------------------------------------------
handle_info(_Info, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: terminate(Reason, State) -> void()
%% Description: This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any necessary
%% cleaning up. When it returns, the gen_server terminates with Reason.
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, State) ->
	CacheId=State#state.cache,
	Factory=State#state.factory,
	CloseFun=Factory#rsc_factory.close_fun,
	Errs=ets:foldl(
		   fun(?ENTRY(RscId,_,_,Handle),AccIn)->
				   try CloseFun(Handle) of
					   _ ->
						   AccIn
				   catch
					   _:Err ->
						   [{close_err, RscId, Err}|AccIn]
				   end
		   end, [], CacheId),
	case Errs of
		[] ->
			ok;
		_ ->
			throw({close_errs,Errs})
	end.

%%--------------------------------------------------------------------
%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
%% Description: Convert process state when code is changed
%%--------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
srv_call(PoolName,Request,ErrTag) ->
	?CHECKERR(gen_server:call(PoolName, Request, infinity),ErrTag).

checkerr({error,Reason},ErrTag) ->
	throw({ErrTag,Reason});
checkerr(Result,_ErrTag) ->
	Result.

%%--------------------------------------------------------------------
%%% Test functions
%%--------------------------------------------------------------------
basic_test_() ->
	{foreach, 
	 local, 
	 fun setup_/0, 
	 fun teardown_/1, 
	 [fun(PoolName)->
			  ?_test(test_open(PoolName))
	  end,
	  fun(PoolName)->
			  ?_test(test_same_open(PoolName))
	  end,
	  fun(PoolName)->
			  ?_assertThrow({open_err,_}, test_bad_open(PoolName))
	  end,
	  fun(PoolName)->
			  ?_test(test_close(PoolName))
	  end,
	  fun(PoolName)->
			  ?_assertThrow({close_err,_}, test_bad_close(PoolName))
	  end,
	  fun(PoolName)->
			  ?_assertThrow({open_err,{pool_full,_,_}}, test_pool_overflow(PoolName))
	  end,
	  fun(PoolName)->
			  ?_test(test_pool_cleanup_prc_exit(PoolName))
	  end,
	  fun(PoolName)->
			  ?_test(test_short_action(PoolName))
	  end,
	  fun(PoolName)->
			  ?_test(test_long_action(PoolName))
	  end
	 ]}.
%% ,
%% 	 fun(PoolName)->
%% 			 [?_test(test_open(PoolName)),
%% 			  ?_assertException(throw, {open_err,_}, test_bad_open(PoolName))]
%% %% 			[?_test(test_open(PoolName)),
%% %% 			 ?_assertException(throw, {open_err,_}, test_bad_open(PoolName)),
%% %% 			 ?_test(test_close(PoolName)),
%% %% 			 ?_test(test_bad_close(PoolName)),
%% %% 			 ?_test(test_pool_overflow(PoolName)),
%% %% 			 ?_test(test_pool_cleanup(PoolName)),
%% %% 			 ?_test(test_short_action(PoolName)),
%% %% 			 ?_test(test_long_action(PoolName))
%% %% 			 ]
%% 	 end}.
cache_id_by_pool(PoolName) ->
	list_to_atom((atom_to_list(PoolName)++"_cache")).
lookup_element(PoolName, Key, Pos) ->
	CacheId=cache_id_by_pool(PoolName),
	ets:lookup_element(CacheId, Key, Pos).
cache_size(PoolName) ->
	CacheId=cache_id_by_pool(PoolName),
	ets:info(CacheId, size).


test_open(PoolName) ->
	open(PoolName, Log={rw,"./test.log"}),
	?assert(lookup_element(PoolName, Log, 3)=:=1),
	Log.
test_same_open(PoolName) ->
	open(PoolName, Log={rw,"./test.log"}),
	open(PoolName, Log),
	?assert(lookup_element(PoolName, Log, 3)=:=2).
test_bad_open(PoolName) ->
	open(PoolName, {rw1,"./test.log"}).

test_close(PoolName) ->
	open(PoolName, Log={rw,"./test.log"}),
	close(PoolName,Log),
	?assert(lookup_element(PoolName, Log, 3)=:=0).

test_bad_close(PoolName) ->
	close(PoolName,{rw1,"./test.log"}).

test_pool_overflow(PoolName) ->
	[open(PoolName, {rw,"./test.log"++[N]})||N<-"1234567890o"].

test_pool_cleanup_prc_exit(PoolName) ->
	Logs=[{rw,"./test.log"++[N]}||N<-"1234567890"],
	ReplyPid=self(),
	WorkPid=spawn_link(fun()->
					   [open(PoolName, Log)||Log<-Logs],
					   ReplyPid ! {opened,self()},
					   receive
						   stop ->
							   ok
					   end
			   end),
	receive
		{opened,WorkPid} ->
			?assert(cache_size(PoolName)=:=10),
			 [?assert(lookup_element(PoolName, Log, 3)=:=1)||Log<-Logs]
	end,
	WorkPid ! stop,
	timer:sleep(100),
	[?assert(lookup_element(PoolName, Log, 3)=:=0)||Log<-Logs].

test_short_action(PoolName) ->
	Log={rw,Filename="./test.log"},
	short_action(
	  PoolName, 
	  Log, 
	  fun(LogPassed) when LogPassed==Log ->
			  ?assert(lookup_element(PoolName, Log, 3)=:=0),
			  disk_log:alog_terms(Log, lists:seq(1,1000))
	  end),
	?assert(lookup_element(PoolName, Log, 3)=:=0),
	disk_log:sync(Log),
	?assert(filelib:file_size(Filename)>1000).

test_long_action(PoolName) ->
	Log={rw,Filename="./test.log"},
	long_action(
	  PoolName, 
	  Log, 
	  fun(LogPassed) when LogPassed==Log ->
 			  ?assert(lookup_element(PoolName, Log, 3)=:=1),
			  disk_log:alog_terms(Log, lists:seq(1,1000))
	  end),
	timer:sleep(100),
	disk_log:sync(Log),
	?assert(lookup_element(PoolName, Log, 3)=:=0),
	?assert(filelib:file_size(Filename)>1000).

setup_() ->
	ReplyPid=self(),
	PoolName=test_disk_log_pool,
	spawn(fun()->
				  ReplyPid ! start_link(PoolName,10,0.5,disk_log_pool:new_factory())
		  end),
	receive
		M ->
			{ok,_Pid}=M
	end,
	PoolName.
teardown_(PoolName) ->
	stop(PoolName),
	lists:foreach(fun file:delete/1, filelib:wildcard("test*log*")).
